package com.hujf.flink.tableSql;

import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.*;
import org.apache.flink.table.api.bridge.java.BatchTableEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.descriptors.TableDescriptor;

import static org.apache.flink.table.api.Expressions.$;

/**
 * @Author hujf
 * @Date 2022/3/25 10:04
 * @Description TODO
 */
public class CommonApiTest {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment environment = StreamExecutionEnvironment.getExecutionEnvironment();
        environment.setParallelism(1);
        EnvironmentSettings settings = EnvironmentSettings.newInstance()
                .inStreamingMode()
                .useBlinkPlanner()
                .build();

        StreamTableEnvironment tableEnvironment1 = StreamTableEnvironment.create(environment);


        String createTble = "CREATE TABLE clickTable (" +
                " user_name STRING , " +
                " url STRING ," +
                "ts BIGINT )" +
                "WITH(" +
                "'connector' = 'filesystem'," +
                " 'path'='file/click.txt'," +
                " 'format' ='csv' )";
        tableEnvironment1.executeSql(createTble);


          Table clickTable = tableEnvironment1.from("clickTable");
          Table select = clickTable.where($("user_name").isEqual("Marry")).select($("user_name"), $("url"));
        //必须创建临时视图
          tableEnvironment1.createTemporaryView("results",select);


        Table select2 = tableEnvironment1.sqlQuery("select url,user_name from clickTable where user_name = 'Marry' ");


//        //创建一张用于输出的表
//        String getTable  = "CREATE TABLE outTable (" +
//                " user_name STRING , " +
//                " url STRING" +
//                ")" +
//                "WITH(" +
//                "'connector' = 'filesystem'," +
//                " 'path'='file/output'," +
//                " 'format' ='csv' )";
//
//        tableEnvironment1.executeSql(getTable);
//
//        //输出表
//
//        select.executeInsert("outTable");

        Table table = tableEnvironment1.sqlQuery("select user_name ,count(url) as cnt from clickTable group by user_name");


        String getTable2 = "CREATE TABLE outTable2 (" +
                " user_name STRING , " +
                " url STRING" +
                ")" +
                "WITH(" +
                "'connector' = 'filesystem'," +
                " 'path'='file/output'," +
                " 'format' ='csv' )";


        String print = "CREATE TABLE printTab (" +
                " user_name STRING , " +
                " url STRING" +
                ")" +
                "WITH(" +
                "'connector' = 'print'" +
               " )";

        String print2 = "CREATE TABLE printTab2 (" +
                " user_name STRING , " +
                " cnt BIGINT" +
                ")" +
                "WITH(" +
                "'connector' = 'print'" +
                " )";


   //     tableEnvironment1.executeSql(getTable2);
     //   tableEnvironment1.executeSql(print);

        tableEnvironment1.executeSql(print2);

//        select2.executeInsert("outTable2");
        table.executeInsert("printTab2");

  //      table.executeInsert("outTable2");


        //执行聚合查询




//        EnvironmentSettings settings1 = EnvironmentSettings.newInstance()
//                .inStreamingMode()
//                .useOldPlanner()
//                .build();
//
//        TableEnvironment tableEnvironment1 = TableEnvironment.create(settings1);
//
//
//        ExecutionEnvironment batchEnv = ExecutionEnvironment.getExecutionEnvironment();
//        BatchTableEnvironment batchTableEnvironment = BatchTableEnvironment.create(batchEnv);
//
//        EnvironmentSettings settings3 = EnvironmentSettings.newInstance()
//                .inBatchMode()
//                .useBlinkPlanner()
//                .build();
//        TableEnvironment tableEnvironment3 = TableEnvironment.create(settings3);
//
//
//        String createTble = "CREATE TABLE clickTable (" +
//                " user_name STRING , " +
//                " url STRING ," +
//                "ts BIGINT )" +
//                "WITH(" +
//                "'connector' = 'filesystem'," +
//                " 'path'='file/click.txt'," +
//                " 'format' ='csv' )";
//
//        //创建一张输出表
//        tableEnvironment1.executeSql(createTble);
//
//
//        //创建一张用于输出的表
//        String getTable  = "CREATE TABLE outTable (" +
//                " user_name STRING , " +
//                " url STRING" +
//                ")" +
//                "WITH(" +
//                "'connector' = 'filesystem'," +
//                " 'path'='file/output'," +
//                " 'format' ='csv' )";
//        tableEnvironment1.executeSql(getTable);

    }
}
